package vn.ducquoc.cassandra;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.List;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TJSONProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

/**
 * Cassandra Thrift example.
 * 
 * @author ducquoc
 */
public class CassandraThriftExample {

  public static final String CASSANDRA_HOST = "192.168.22.210";
  public static final int CASSANDRA_PORT = 9160;

  public static final String SAMPLE_KEYSPACE = "prstest";
  public static final String SAMPLE_COLUMN_FAMILY = "ductest";

  static final long TIMESTAMP = 1010102011000L;

  public static void main(String[] args) throws Exception {
    TTransport tr = new TFramedTransport(new TSocket(CASSANDRA_HOST,
            CASSANDRA_PORT));
    TProtocol proto = new TJSONProtocol(tr);
    Cassandra.Client client = new Cassandra.Client(proto);
    tr.open();

    String key_user_id = "1st_rowKey";

    // insert data
    long timestamp = 1300000010102011L;//System.currentTimeMillis();
    client.set_keyspace(SAMPLE_KEYSPACE);
    ColumnParent parent = new ColumnParent(SAMPLE_COLUMN_FAMILY);

    Column nameColumn = new Column(toByteBuffer("name"));
    nameColumn.setValue(toByteBuffer("Duc Quoc"));
    nameColumn.setTimestamp(timestamp);
    client.insert(toByteBuffer(key_user_id), parent, nameColumn,
            ConsistencyLevel.ONE);

    Column ageColumn = new Column(toByteBuffer("age"));
    ageColumn.setValue(toByteBuffer("24"));
    ageColumn.setTimestamp(timestamp);
    client.insert(toByteBuffer(key_user_id), parent, ageColumn,
            ConsistencyLevel.ONE);

    ColumnPath path = new ColumnPath(SAMPLE_COLUMN_FAMILY);

    // read single column
    path.setColumn(toByteBuffer("name"));
    System.out.println(client.get(toByteBuffer(key_user_id), path,
            ConsistencyLevel.ONE));

    // read entire row
    SlicePredicate predicate = new SlicePredicate();
    SliceRange sliceRange = new SliceRange(toByteBuffer(""), toByteBuffer(""),
            false, 10);
    predicate.setSlice_range(sliceRange);

    List<ColumnOrSuperColumn> results = client.get_slice(
            toByteBuffer(key_user_id), parent, predicate, ConsistencyLevel.ONE);
    for (ColumnOrSuperColumn result : results) {
      Column column = result.column;
      System.out.println(toString(column.name) + " -> "
              + toString(column.value));
    }

    tr.close();
  }

  public static ByteBuffer toByteBuffer(String value)
          throws UnsupportedEncodingException {
    return ByteBuffer.wrap(value.getBytes("UTF-8"));
  }

  public static String toString(ByteBuffer buffer)
          throws UnsupportedEncodingException {
    byte[] bytes = new byte[buffer.remaining()];
    buffer.get(bytes);
    return new String(bytes, "UTF-8");
  }

}
